package org.shuma.goods.utils;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.shuma.goods.contant.DelayLevel;
import org.shuma.goods.contant.GoodsConstant;
import org.shuma.goods.domain.entity.ProductBasicInfo;
import org.shuma.goods.mapper.ProductBasicInfoMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author : liuming
 * @date : 2024/6/18 下午7:14
 */
@Component
@Slf4j
@RocketMQMessageListener(
        topic = "product_schedule",
        consumerGroup = "product_schedule_consumer"
)
public class ProductScheduleMessageMng implements RocketMQListener<String> {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private ProductBasicInfoMapper productBasicInfoMapper;

    public void SendProductUpOrDownScheduleInfo(ProductScheduleMessage message) {
        String topic = "product_schedule";
        Gson gson = new Gson();
        String json = gson.toJson(message, ProductScheduleMessage.class);
//        String message = "我是一条异步延迟文本消息:asyncSendStringMessageWithBuilderDelayLevel";
        Message<String> msg = MessageBuilder.withPayload(json)
                // 设置消息类型
                .setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
                .build();
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("productId: {},{}失败{}", message.getDelayLevel(), message.getOperation().getDescription(), sendResult);
            }

            @Override
            public void onException(Throwable e) {
                // 异步发送失败的回调逻辑
                log.error("productId: {},{}失败{}", message.getProductId(), message.getOperation().getDescription(), e.getMessage());
            }
        }, 3000, message.getDelayLevel().getLevel());
    }

    // 每两小时执行一次
    @Scheduled(cron = "0 0 */2 * * ?")
    public void dailyJob() {
        // 创建一个线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
        LambdaQueryWrapper<ProductBasicInfo> queryWrapper = new LambdaQueryWrapper<>();
//        获取所有 达到 确认上架的商品+上线时间<=2小时的商品
        queryWrapper.eq(ProductBasicInfo::getProductCheckState, GoodsConstant.CONFIRM_ONLINE);
        queryWrapper.le(ProductBasicInfo::getTimeOn, System.currentTimeMillis() + DelayLevel.LEVEL_2H.getDelayTime(TimeUnit.MILLISECONDS));
        List<ProductBasicInfo> productList = productBasicInfoMapper.selectList(queryWrapper);
//        将发布上架消息的任务添加到线程池
        for (ProductBasicInfo productBasicInfo : productList) {
            // 为每个产品创建一个Runnable任务
            Runnable sendMessageTask = () -> {
                // 计算延迟时间
                long delayTime = productBasicInfo.getTimeOn().getTime() - new Date().getTime();
                // 创建上架消息
                ProductScheduleMessage message = new ProductScheduleMessage(productBasicInfo.getProductId(), ProductScheduleStatus.UP, delayTime);
                // 发送消息到RocketMQ
                ProductScheduleMessageMng productScheduleMessageMng = new ProductScheduleMessageMng();
                productScheduleMessageMng.SendProductUpOrDownScheduleInfo(message);
            };
            // 将任务提交给线程池执行
            executor.execute(sendMessageTask);
        }
//        获取所有剩余下架时间<=2小时的商品
        LambdaQueryWrapper<ProductBasicInfo> queryWrapper2 = new LambdaQueryWrapper<>();
        queryWrapper2.eq(ProductBasicInfo::getProductCheckState, GoodsConstant.GO_ONLINE);
        queryWrapper2.le(ProductBasicInfo::getTimeOff, System.currentTimeMillis() + DelayLevel.LEVEL_2H.getDelayTime(TimeUnit.MILLISECONDS));
        List<ProductBasicInfo> productList2 = productBasicInfoMapper.selectList(queryWrapper2);
//        将发布下架消息信息的任务添加到线程池
        for (ProductBasicInfo productBasicInfo : productList2) {
            // 为每个产品创建一个Runnable任务
            Runnable sendMessageTask = () -> {
                // 计算延迟时间
                long delayTime = productBasicInfo.getTimeOff().getTime() - new Date().getTime();
                // 创建消息
                ProductScheduleMessage message2 = new ProductScheduleMessage(productBasicInfo.getProductId(), ProductScheduleStatus.DOWN, delayTime);
                // 发送消息到RocketMQ
                ProductScheduleMessageMng productScheduleMessageMng = new ProductScheduleMessageMng();
                productScheduleMessageMng.SendProductUpOrDownScheduleInfo(message2);
            };
        }
        // 关闭线程池，不再接受新任务，但已提交的任务会继续执行
        executor.shutdown();
        // 如果需要等待所有任务执行完毕再继续
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow(); // 等待超时后强制关闭
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt(); // 重新设置中断状态
        }
    }

    //    消息商品上下架信息
    @Override
    public void onMessage(String message) {
        Gson gson = new Gson();
        ProductScheduleMessage productScheduleMessage = gson.fromJson(message, ProductScheduleMessage.class);
//        消息上架信息
        if (productScheduleMessage.getOperation().equals(ProductScheduleStatus.UP)) {
            ProductBasicInfo productBasicInfo = productBasicInfoMapper.selectById(productScheduleMessage.getProductId());
            productBasicInfo.setProductCheckState(GoodsConstant.GO_ONLINE);
        }
//        消息下架信息
        if (productScheduleMessage.getOperation().equals(ProductScheduleStatus.DOWN)) {
            ProductBasicInfo productBasicInfo = productBasicInfoMapper.selectById(productScheduleMessage.getProductId());
            productBasicInfo.setProductCheckState(GoodsConstant.BE_OFFLINE);
        }
    }
}
